package com.example.spring.pulsar.demo.controller;


import com.example.spring.pulsar.demo.domain.so.UserSo;
import com.wu.framework.inner.layer.web.EasyController;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.springframework.boot.CommandLineRunner;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;

@EasyController("/pulsar")
public class PulsarController implements CommandLineRunner {

    private final PulsarClient pulsarClient;

    public PulsarController(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
    }

    /**
     * @description: 推送数据到主题
     * @author: 吴佳伟
     * @date: 2021/5/20
     * @param: null
     * @return:
     */
    @PostMapping("/add/{topic}")
    public MessageId add(@PathVariable String topic, @RequestParam String message) throws PulsarClientException {
        return pulsarClient.newProducer().
                topic(topic).create().send(message.getBytes(StandardCharsets.UTF_8));
    }

    //    @PostConstruct
    public void save() throws PulsarClientException {
        Producer<byte[]> producer = pulsarClient.newProducer().topic("more").create();
        for (int i = 0; i < 1000; i++) {
            new Thread(() -> {
                while (true) {
                    try {
                        producer.send(("当前时间" + LocalDate.now()).getBytes(StandardCharsets.UTF_8));
                    } catch (PulsarClientException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

    /**
     * @description: 监听主题数据
     * @author: 吴佳伟
     * @date: 2021/5/20
     * @param: null
     * @return:
     */
    @PostMapping("/monitor/{topic}")
    public String monitor(@PathVariable String topic) throws PulsarClientException {
        Consumer<byte[]> subscribe = pulsarClient.newConsumer().topic(topic).subscriptionName(topic).subscribe();
        new Thread(() -> {
            while (true) {
                try {
                    Message<byte[]> receive = subscribe.receive();
                    System.out.println("接收到的数据:" + receive);
                    subscribe.acknowledge(receive);
                } catch (PulsarClientException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        return "启动一个线程监听主题:" + topic;
    }


    /**
     * @description: 发送实体到指定主题
     * @author: 吴佳伟
     * @date: 2021/5/20
     */
    @PostMapping("/add/schema/{topic}")
    public MessageId addSchema(@PathVariable String topic, @RequestBody UserSo userSo) throws PulsarClientException {
        return pulsarClient.newProducer(JSONSchema.of(UserSo.class)).
                topic(topic).create().send(userSo);
    }

    /**
     * @description: 监听指定主题并将数据转换成指定schema类型的实体数据
     * @author: 吴佳伟
     * @date: 2021/5/20
     * @param: null
     * @return:
     */
    @PostMapping("/monitor/schema/{topic}")
    public String monitorSchema(@PathVariable String topic) throws PulsarClientException {
        Consumer<UserSo> subscribe = pulsarClient.newConsumer(JSONSchema.of(UserSo.class)).topic(topic).subscriptionName(topic).subscribe();
        new Thread(() -> {
            while (true) {
                try {
                    Message<UserSo> receive = subscribe.receive();
                    System.out.println("接收到的数据:" + receive);
                    subscribe.acknowledge(receive);
                } catch (PulsarClientException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        return "启动一个线程监听主题:" + topic;
    }


    @Override
    public void run(String... args) throws Exception {
        Producer<UserSo> userSoProducer = pulsarClient.newProducer(JSONSchema.of(UserSo.class)).
                topic("schema-topic").create();
        UserSo userSo = new UserSo();
        userSo.setAge(10l);
        userSo.setName("user");
        userSo.setAddress("address");
        userSoProducer.send(userSo);

//        Consumer<UserSo> subscribe = pulsarClient.newConsumer(JSONSchema.of(UserSo.class)).
//                topic("schema-topic").subscriptionName("schema-topic").subscribe();
//        Message<UserSo> receive = subscribe.receive();
//        System.out.println(receive.getValue());
    }





}
